#include "discovery/discovery_service.h"
#include <string>
#include <thread>
#include <unistd.h>

#include "boost/asio.hpp"
#include "log/logger.h"

#include "pub_sub/pub_sub_infos.h"

DiscoveryService::DiscoveryService(boost::asio::io_context &io_context,
                                   const std::string &service_name)
    : io_context_(io_context), service_name_(service_name), socket_(io_context),
      timer_(io_context) {
  if (service_name.empty()) {
    service_name_ = "default_service_" + std::to_string(getpid());
  }
  interval_ = 1;

  boost::asio::ip::udp::endpoint endpoint(boost::asio::ip::udp::v4(), 17240);
  socket_.open(endpoint.protocol());

  socket_.set_option(boost::asio::socket_base::broadcast(true));

  socket_.set_option(boost::asio::socket_base::reuse_address(true));

  socket_.bind(endpoint);

  do_receive();
  timer_to_send_broadcast();
  timer_to_broadcast_pub_sub_infos();
}

void DiscoveryService::send_broadcast(const std::string &message) {
  boost::asio::ip::udp::endpoint broadcast_endpoint(
      boost::asio::ip::address_v4::broadcast(), // 255.255.255.255
      17240);

  socket_.async_send_to(
      boost::asio::buffer(message), broadcast_endpoint,
      [this, message](boost::system::error_code ec, std::size_t bytes_sent) {
        if (!ec) {
          JC_INFO("Broadcast sent: {} ({} bytes)", message, bytes_sent);
        } else {
          JC_ERROR("Failed to send broadcast: {}", ec.message());
        }
      });
}

DiscoveryService::~DiscoveryService() { socket_.close(); }

void DiscoveryService::handle_discovery_message(const std::string &message) {
  static std::string pid_str = std::to_string(getpid());
  if (message.find(pid_str) != std::string::npos)
    return;
  JC_INFO("Received discovery message: {}", message);
}

void DiscoveryService::do_receive() {
  socket_.async_receive_from(
      boost::asio::buffer(buffer_), sender_endpoint_,
      [this](boost::system::error_code ec, std::size_t bytes_received) {
        if (!ec && bytes_received > 0) {
          handle_discovery_message(std::string(buffer_.data(), bytes_received));
        }
        do_receive();
      });
}

void DiscoveryService::timer_to_send_broadcast() {
  timer_.expires_after(std::chrono::seconds(interval_));
  timer_.async_wait([this](const boost::system::error_code &ec) {
    if (!ec) {
      send_broadcast(service_name_);
    }
    timer_to_send_broadcast();
  });
}

void DiscoveryService::timer_to_broadcast_pub_sub_infos() {
  timer_.expires_after(std::chrono::seconds(interval_));
  timer_.async_wait([this](const boost::system::error_code &ec) {
    if (!ec) {
      send_broadcast(pub_sub_infos_to_string());
    }
    timer_to_broadcast_pub_sub_infos();
  });
}